package com.yuninglong.demo2;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class StreamingJob {

    public static void main(String[] args) throws Exception{
        //1 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2 设置监听的socket网络端口
        DataStream<String> text = env.socketTextStream("192.168.202.10", 9000);

        //处理数据
        FlatMapFunction<String, Tuple2<String, Integer>> flatMapper = new FlatMapFunction<String, Tuple2<String, Integer>>() {
            //序列化
            private static final long serialVersionUID = 1L;

            @Override
            public void flatMap(String arg0, Collector<Tuple2<String, Integer>> collector) throws Exception {
                //3 以空格分割数据
                String[] strs = arg0.toLowerCase().split(" ");

                //4 遍历 收集数据
                for (String str : strs) {
                    if (str.length() > 0) {
                        collector.collect(new Tuple2<String, Integer>(str, 1));
                    }
                }
            }

        };

        //5 处理数据
        DataStream<Tuple2<String, Integer>> dataStream = text.flatMap(flatMapper)
                .keyBy(1)
                .sum(1);

        //6 打印数据
        dataStream.print();

        SerializationSchema schema=new SerializationSchema() {
            @Override
            public byte[] serialize(Object o) {
                return new byte[0];
            }
        };
        dataStream.writeToSocket("192.168.202.10",9100,schema);

        //7 启动项目
        env.execute("启动项目");

    }

}
